package com.example.kafka.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.consumer.OffsetCommitCallback;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.Collections;
import java.util.Map;

/**
 * @application:
 * @author: LiYuhang
 * @Date: 2021/6/26 14:45
 * @version: 0.1
 * @Description :
 * @Modified By:
 */
@Slf4j
public class ConsumerDemo2 implements Consumer {

    private final KafkaConsumer<String, String> consumer;
    private final String topic;


    public ConsumerDemo2(String topic,String groupID) {
        this.consumer = new KafkaConsumer<String, String>(Consumer.getDefaultConfig(groupID));
        this.topic = topic;
    }

    public void run() {
        this.consumer.subscribe(Collections.singleton(this.topic));
        try {
            while (true) {
                ConsumerRecords<String, String> records =
                        consumer.poll(Duration.ofMillis(100));
                if (records.isEmpty()) {
                    log.warn("未读取到消息，sleep 500");
                    try {
                        Thread.sleep(500);
                    } catch (InterruptedException e) {

                    }
                } else {
                    for (ConsumerRecord<String, String> record : records) {
                        log.info("run:{},{}",record.value(),record);
                    }
                    consumer.commitAsync(new OffsetCommitCallback() {
                        @Override
                        public void onComplete(Map<TopicPartition, OffsetAndMetadata> map,Exception e) {
                            log.info("commitAsync map :{}",map);
                            if(e!=null){
                                log.error("当前错误信息时提交的偏移量:{},msg:{}",map,e.getMessage(),e);
                            }
                        }
                    });
                }
            }
        }catch (Exception e){
            log.error("error  msg:{}",e.getMessage(),e);
        }finally {
            try {
                consumer.commitSync();    //同步提交
            }finally {
                consumer.close();
            }
        }

    }

    public static void main(String[] args) {
        //test
        //"topic.group3"
        ConsumerDemo2 demo1 = new ConsumerDemo2("test","topic.group1");
        demo1.run();
    }


}
